package agent

import (
	"encoding/json"
	"fmt"
	"gopkg.in/vmihailenco/msgpack.v2"
	"logagent/config"
	"logagent/rds"
	"logagent/register"
	"logagent/taillog"
	"logagent/utils"
	"os"
	"os/signal"
	"strings"
	"sync"
	"syscall"
	"time"
)

type TailAgent struct {
	wg   sync.WaitGroup
	mgr  *taillog.TailObjMgr
	work []chan struct{}
}

func (t *TailAgent) Run() {
	// 1. 初始化日志
	utils.Init()
	// 2. 配置初始化
	config.Init()
	// 3. 初始化rds
	rds.Init(config.Cfg.RdsConf.Hosts, config.Cfg.RdsConf.DB, config.Cfg.RdsConf.Pwd)
	// 4. taillog 初始化
	register.Init(config.Cfg.SinceDb)
	mgr, err := taillog.InitTail(config.Cfg.Collect, config.Cfg.ChanSize)
	if err != nil {
		utils.Fatalf("无法初始化taillog")
	}
	t.mgr = mgr
	// 5. 开启服务
	t.serve()
	startTime := time.Now()
	fmt.Println("程序启动成功,当前时间: ", startTime.Format("2006-01-02 15:04:05"))
	// 6. 开启监听
	t.signal()
	// 7. 等待关闭
	t.Close()
	endTime := time.Now()
	fmt.Println("程序结束,运行时长: ", endTime.Sub(startTime).Round(1000000))

}

func (t *TailAgent) serve() error {
	// 创建redis连接
	client := rds.GetRds()
	for i := 0; i < 5; i++ {
		t.wg.Add(1)
		exitChan := make(chan struct{})
		go func(i int) {
			defer t.wg.Done()
			ch := t.mgr.GetChannel()
		FOR:
			for true {
				select {
				case msg, ok := <-ch:
					if !ok {
						break FOR
					}
					if len(msg.Msg) == 0 {
						continue
					}
					key := msg.Key
					pos := msg.Pos
					msg.Key = ""
					msg.Pos = 0
					msg.Msg = strings.TrimSpace(msg.Msg)
					bytes, _ := msgpack.Marshal(msg)
					e := client.LPush(key, string(bytes)).Err()
					if e != nil {
						fmt.Println(e)
						time.Sleep(time.Second)
						continue
					}
					// 存储到Register
					register.Reg.Mutex.Lock()
					register.Reg.Mreg[msg.File] = register.Record{
						Path: msg.File,
						Pos:  pos,
					}
					register.Reg.Mutex.Unlock()
					utils.Println("current goroutine is :", i)
					s, _ := json.Marshal(msg)
					utils.Println("line content: ", string(s))
				case <-exitChan:
					break FOR
				}
			}
		}(i)
		t.work = append(t.work, exitChan)
	}
	return nil
}

func (t *TailAgent) signal() {
	signals := make(chan os.Signal)
	signal.Notify(signals, syscall.SIGINT, syscall.SIGTERM)
	go func() {
		select {
		case <-signals:
			// 将所有的子goroutine结束
			t.mgr.Close()
			for _, exit := range t.work {
				exit <- struct{}{}
			}
		}
	}()

}

func (t *TailAgent) Close() {
	t.wg.Wait()
}
